package com.study.flink.window

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * Event Time Session Window
  * 相邻两次数据的EventTime的时间差超过指定的时间间隔就会触发执行。
  * @author stephen
  * @date 2019-07-22 17:29
  */
object ScalaEventTimeSessionWindowDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置为Event Time，默认为Processing Time
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    // 2 获取输入数据
    import org.apache.flink.api.scala._
    // a 1547863262000
    // a 1547863272000
    // a 1547863277000
    // a 1547863282000
    // a 1547863292000
    val dataStream = env.socketTextStream("localhost",9999)

    // 3 Transformation
    val resultStream = dataStream
      .map(x => {
        val arr = x.split(" ")
        // word 时间戳 次数
        (arr(0), arr(1).toLong, 1)
      })
      // watermark配置，设置最大容忍乱序时间为1秒
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.seconds(1)) {
      override def extractTimestamp(element: (String, Long, Int)): Long = {
        element._2
      }
    })
      .keyBy(_._1)
      // Session窗口：两次EventTime超过8秒触发执行
      .window(EventTimeSessionWindows.withGap(Time.seconds(8)))
      .reduce((x, y) => (x._1, Math.max(x._2, y._2), x._3 + y._3))

    // 4 输出
    resultStream.print().setParallelism(1)

    // 5 启动任务
    env.execute("Event Time Session Window")
  }
}
